package com.blue.Impl.system;

import com.alibaba.fastjson.JSON;
import com.blue.common.constant.CommonConstant;
import com.blue.common.exception.CustomException;
import com.blue.mapper.system.DataSyncConfMapper;
import com.blue.mapper.system.GenerateMapper;
import com.blue.po.system.DataSyncConfPO;
import com.blue.po.system.DataSyncInfoPO;
import com.blue.po.system.SyncFailDetailPO;
import com.blue.service.system.IDataSyncInfoService;
import com.blue.service.system.IDataSyncService;
import com.blue.service.system.ISyncFailDetailService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static com.baomidou.mybatisplus.core.toolkit.StringPool.ASTERISK;

@Service
@Slf4j
public class DataSyncServiceImpl implements IDataSyncService {

    @Autowired
    private IDataSyncInfoService dataSyncInfoService;

    @Autowired
    private ISyncFailDetailService syncFailDetailService;

    @Autowired
    private GenerateMapper generateMapper;

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private DataSyncConfMapper dataSyncConfMapper;

    private static final String regex = "\\$\\{([^}]+)\\}";

    @Override
    @Async("customExecutor")
    public void dataSync(DataSyncConfPO dataSyncConfPO) {
        String beginTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
        log.info("{}开始同步{}配置的数据", beginTime, dataSyncConfPO.getConfNm());
        DataSyncInfoPO dataSyncInfoPO = new DataSyncInfoPO();
        dataSyncInfoPO.setHasFail(CommonConstant.NO);
        BigDecimal failNum = BigDecimal.ZERO;
        String date = new SimpleDateFormat("yyyy-MM-dd").format(new Date());
        dataSyncInfoPO.setSyncDate(date);
        try {
            //拼接查询sql
            String tableNm = dataSyncConfPO.getTableNm();
            if (StringUtils.isBlank(tableNm)) {
                throw new CustomException("表名为空");
            }
            String whereCondition = replaceWhereCondition(dataSyncConfPO.getWhereCondition());
            String syncField = dataSyncConfPO.getSyncField();
            String onlyField = dataSyncConfPO.getOnlyField();
            List<String> fieldList;
            //同步字段为*表示同步表全部字段
            if (ASTERISK.equals(syncField)) {
                List<Map<String, String>> ColumnsMapList = generateMapper.queryColumns(tableNm);
                fieldList = ColumnsMapList.stream().map(item -> item.get("columnName")).toList();
            } else {
                fieldList = Arrays.asList(syncField.split(","));
            }
            String fieldStr = String.join(",", fieldList);
            //检查同步字段和条件
            checkConf(whereCondition, fieldStr, tableNm);
            //获取数据总数量
            String countSql = "select count(id) from " + tableNm + " where " + whereCondition;
            log.info("{}同步数据配置countSql为:{}", dataSyncConfPO.getConfNm(), countSql);
            long count = dataSyncConfMapper.getCount(countSql);
            //分页同步每次查询2000条数据
            int pageNum = 1;
            if (count > 2000) {
                pageNum = (int) Math.ceil((double) count / 2000);
            }
            log.info("{}同步数据配置pageNum为:{}", dataSyncConfPO.getConfNm(), pageNum);
            for (int i = 1; i <= pageNum; i++) {
                //分页查询需要同步的数据
                String selectSql = "select " + fieldStr + " from " + tableNm + " where " +
                        whereCondition + " limit " + (i - 1) * 2000 + " , 2000";
                log.info("{}同步数据配置selectSql为:{}", dataSyncConfPO.getConfNm(), selectSql);
                List<Map<String, Object>> dataMapList = dataSyncConfMapper.selectData(selectSql);
                dataSyncInfoPO.setDataNum(BigDecimal.valueOf(dataMapList.size()));
                log.info("{}同步数据配置开始发送MQ,本次同步数据条数为:{}", dataSyncConfPO.getConfNm(), dataMapList.size());
                for (Map<String, Object> dataMap : dataMapList) {
                    Object onlyFieldValue = null;
                    Map<String, String> syncMap = new HashMap<>();
                    try {
                        //拼接插入sql
                        List<Object> valueList = new ArrayList<>();
                        for (String field : fieldList) {
                            Object value = dataMap.get(field);
                            if (onlyField != null && onlyField.equals(field)) {
                                if (value instanceof String) {
                                    onlyFieldValue = "'" + value + "'";
                                } else {
                                    onlyFieldValue = value;
                                }
                            }
                            valueList.add(value);
                        }
                        String values = valueList.stream().map(item -> {
                            if (item == null) {
                                item = "null";
                            } else if (item instanceof String) {
                                item = "'" + item + "'";
                            }
                            return item.toString();
                        }).collect(Collectors.joining(","));
                        String insertSql = "insert into " + tableNm + " ( " + fieldStr + " ) values ( " + values + " )";
                        syncMap.put("insertSql", insertSql);
                        //拼接删除sql
                        if (onlyFieldValue != null) {
                            String deleteSql = "delete from " + tableNm + " where " + onlyField + "=" + onlyFieldValue;
                            syncMap.put("deleteSql", deleteSql);
                        }
                        //同步数据
                        syncMap.put("confId", dataSyncConfPO.getId());
                        rabbitTemplate.convertAndSend("dataSyncExchange", "dataSyncKey", JSON.toJSONString(syncMap));
                    } catch (Exception e) {
                        failNum = failNum.add(BigDecimal.ONE);
                        dataSyncInfoPO.setHasFail(CommonConstant.YES);
                        SyncFailDetailPO syncFailDetailPO = new SyncFailDetailPO();
                        syncFailDetailPO.setSyncDate(date);
                        syncFailDetailPO.setFailDes(e.getMessage());
                        syncFailDetailPO.setConfId(dataSyncConfPO.getId());
                        syncFailDetailPO.setFailTime(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                        syncFailDetailPO.setFailData(JSON.toJSONString(syncMap));
                        syncFailDetailService.save(syncFailDetailPO);
                    }
                }
                dataSyncInfoPO.setStatus(CommonConstant.SYNC_SUCCESS);
            }
        } catch (Exception e) {
            dataSyncInfoPO.setStatus(CommonConstant.SYNC_FAIL);
            dataSyncInfoPO.setFailDes(e.getMessage());
            log.error("{}同步数据配置同步失败,异常：{}", dataSyncConfPO.getConfNm(), e.getMessage());
        } finally {
            dataSyncInfoPO.setConfId(dataSyncConfPO.getId());
            dataSyncInfoPO.setBeginTime(beginTime);
            dataSyncInfoPO.setFailNum(failNum);
            String endTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
            dataSyncInfoPO.setEndTime(endTime);
            dataSyncInfoService.save(dataSyncInfoPO);
            log.info("{}{}配置数据同步结束", endTime, dataSyncConfPO.getConfNm());
        }
    }

    //替换where条件时间函数
    private String replaceWhereCondition(String whereCondition) {
        if (StringUtils.isBlank(whereCondition)) {
            return "1 = 1";
        }
        //替换条件中的时间函数${函数:格式}
        Pattern pattern = Pattern.compile(regex);
        Matcher matcher = pattern.matcher(whereCondition);
        while (matcher.find()) {
            String value = matcher.group(1);
            log.info("查询条件中存在时间函数:{}", value);
            String[] valueArr = value.split(":");
            if (valueArr.length != 2) {
                throw new CustomException("时间函数定义错误");
            }
            String var = valueArr[0];
            Calendar calendar = Calendar.getInstance();
            switch (var) {
                case "date", "month", "year":
                    break;
                case "yesterday":
                    calendar.add(Calendar.DATE, -1);
                    break;
                case "lastWeek":
                    calendar.add(Calendar.WEEK_OF_YEAR, -7);
                    break;
                case "lastMonth":
                    calendar.add(Calendar.MONTH, -1);
                    break;
                case "lastYear":
                    calendar.add(Calendar.YEAR, -1);
                    break;
                default:
                    throw new CustomException("时间函数参数有误");
            }
            String dateFormat = valueArr[1];
            SimpleDateFormat format = new SimpleDateFormat(dateFormat);
            String date = "'" + format.format(calendar.getTime()) + "'";
            whereCondition = whereCondition.replace("${" + value + "}", date);
            log.info("替换{}时间函数后条件:{}", value, whereCondition);
        }
        return whereCondition;
    }

    //检查同步字段和条件配置是否正确
    private void checkConf(String whereCondition, String field, String tableNm) {
        if (StringUtils.isBlank(field)) {
            throw new CustomException("同步字段不能为空");
        }
        String checkSql = "select " + field + " from " + tableNm + " where " + whereCondition + " limit 1";
        try {
            dataSyncConfMapper.selectData(checkSql);
        } catch (Exception e) {
            throw new CustomException("同步字段或条件有误");
        }
    }
}
